package com.hyts.reactor.handler.subcribe;

import java.util.Arrays;
import java.util.Iterator;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * project-name:coffice-toolbox
 * package-name:com.hyts.reactor.handler.subcribe
 * author:Libo/Alex
 * create-date:2019-06-25 15:42
 * email:liboware@gmail.com
 * description:此类主要用于: 数组模式方式的处理方式
 */
public class ArraySubscription<T> implements Subscription {

    /**
     * 订阅者操作
     */
    private Subscriber<? super T> subscriber;

    /**
     * 操作元素
     */
    private final T[] arrayElement;

    /**
     * 构造器
     * @param subscriber
     * @param arrayElement
     */
    public ArraySubscription(Subscriber<? super T> subscriber, T[] arrayElement) {
        this.subscriber = subscriber;
        this.arrayElement = arrayElement;
    }

    private AtomicBoolean cancel = new AtomicBoolean(false);

    /**
     * 请求操作的处理元素个数
     * @param n the strictly positive number of elements to requests to the upstream
     */
    @Override
    public void request(long n) {
        //判断是否取消
        if(cancel.get()){
            return ;
        }
        long length = arrayElement == null ? 0 : arrayElement.length;
        //进行遍历操作服务
        Iterator<T> iterable = Arrays.asList(arrayElement).listIterator();
        while(cancel.get() && iterable.hasNext()){
            subscriber.onNext(iterable.next());
        }
        subscriber.onComplete();
    }

    @Override
    public void cancel() {
        cancel.compareAndSet(true,false);
    }
}
